package cn.ac.iie.di.ban.ycl.cleaner;

import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.ban.ycl.cleaner.handler.CleanerHandler;
import cn.ac.iie.di.ban.ycl.cleaner.handler.KillHandler;
import cn.ac.iie.di.ban.ycl.cleaner.receiver.MessageFailReceiver;
import cn.ac.iie.di.ban.ycl.cleaner.receiver.MessageSuccessReceiver;
import cn.ac.iie.di.ban.ycl.cleaner.sender.DeduplicationMessageSenderThread;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.configuration2.FileBasedConfiguration;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.configuration2.builder.FileBasedConfigurationBuilder;
import org.apache.commons.configuration2.builder.fluent.Parameters;
import org.apache.commons.configuration2.convert.DefaultListDelimiterHandler;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-9-8 14:32:24
 * @version V1.0
 * @desc 程序入口
 */
public class CleanerMain {

	private static final Logger LOG = LoggerFactory.getLogger(CleanerHandler.class);
	public static LinkedBlockingQueue<REMessage> handlerQueue;//接收队列
	public static LinkedBlockingQueue<REMessage> deduplicationQueue;//去重队列
	static final String PROPERTIES_PATH;

	static {
		PROPERTIES_PATH = ClassLoader.getSystemClassLoader()
				.getResource("configuration.properties").getFile();
		if (PROPERTIES_PATH == null || PROPERTIES_PATH.isEmpty()) {
			throw new RuntimeException("properties path is empty!");
		} else {
			LOG.info("PROPERTIES_PATH = {}", PROPERTIES_PATH);
		}
	}

	public static void main(String[] args) {
		try {
			//读取配置文件
			FileBasedConfiguration configuration = initConfiguration();
			//启动数据发送
			deduplicationQueue = new LinkedBlockingQueue<>(50);
			Thread deduplicationSendThread = new Thread(
					new DeduplicationMessageSenderThread(configuration),
					"deduplication send thread");
			deduplicationSendThread.start();

			//启动数据清洗
			int handlerQueueSize = configuration.getInt(Constants.HANDLER_QUEUE_SIZE, 10);
			int handlerThreadNum = configuration.getInt(Constants.HANDLER_THREAD_NUM, 4);

			LOG.info("--------------------------------------------");
			LOG.info("   handler queue size = " + handlerQueueSize);
			LOG.info("   handler thread num = " + handlerThreadNum);
			LOG.info("--------------------------------------------");

			handlerQueue = new LinkedBlockingQueue<>(handlerQueueSize);
			for (int i = 0; i < handlerThreadNum; i++) {
				Thread thread = new Thread(new CleanerHandler(configuration, handlerQueue));
				thread.start();
			}

			//启动数据接收
			String mqNameserver = configuration.getString(Constants.RECEIVER_MQ_NAMESERVER);
			String mqTopic = configuration.getString(Constants.RECEIVER_MQ_READ_TOPIC);
			String mqGroup = configuration.getString(Constants.RECEIVER_MQ_GROUP);
			int consumerThreadNum = configuration.getInt(Constants.RECEIVER_THREAD_NUM, 4);
			LOG.info("--------------------------------------------");
			LOG.info("   receiver mq nameserver = " + mqNameserver);
			LOG.info("        receiver mq topic = " + mqTopic);
			LOG.info("       receiver  mq group = " + mqGroup);
			LOG.info("      receiver thread num = " + consumerThreadNum);
			LOG.info("--------------------------------------------");
			REConnection conn = new REConnection(mqNameserver);
			REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn
					.getReceiveSessionBuilder(mqTopic);

			if (args.length == 0 || true) {
				LOG.info("consumer from last offset!");
				builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
			} else {
				builder.setConsumPosition(ConsumePosition.CONSUME_FROM_TIMESTAMP);
				SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				Date date = sdf.parse(args[0]);
				LOG.info("consumer from date {} ", args[0]);
				builder.setConsumeTimestamp(date.getTime());
			}
			builder.setConsumeThreadNum(consumerThreadNum);
			builder.setHandler(new MessageSuccessReceiver(handlerQueue));
			builder.setFailureHandler(new MessageFailReceiver());
			builder.setGroupName(mqGroup);
			REReceiveSession session = (REReceiveSession) builder.build();
			session.start();

			//程序关闭
			KillHandler killHandler = new KillHandler();
			killHandler.registerSignal("TERM");
			killHandler.putQueue("deduplication queue", deduplicationQueue);
			killHandler.setConn(conn);
			LOG.info("cleaner start success!");
		} catch (ConfigurationException | REConnectionException | ParseException ex) {
			LOG.error("start program error!", ex);
			System.exit(1);
		}
	}

	private static FileBasedConfiguration initConfiguration() throws ConfigurationException {
		Parameters params = new Parameters();
		FileBasedConfigurationBuilder<FileBasedConfiguration> confBuilder
				= new FileBasedConfigurationBuilder<FileBasedConfiguration>(
						PropertiesConfiguration.class)
				.configure(params.properties()
						.setFileName(PROPERTIES_PATH)
						.setListDelimiterHandler(new DefaultListDelimiterHandler(','))
						.setThrowExceptionOnMissing(true));
		FileBasedConfiguration configuration = confBuilder.getConfiguration();
		return configuration;
	}
}
